package vertxRestZeebeKafka;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import io.vertx.core.json.JsonObject;
import io.vertx.kafka.client.consumer.KafkaConsumer;
import io.vertx.kafka.client.producer.KafkaProducer;
import io.vertx.kafka.client.serialization.JsonObjectDeserializer;
import io.vertx.kafka.client.serialization.JsonObjectSerializer;
import io.zeebe.client.ZeebeClient;
import io.zeebe.client.api.response.WorkflowInstanceEvent;
import io.zeebe.client.api.worker.JobWorker;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.LongSerializer;

import java.util.Arrays;
import java.util.List;
import java.util.Properties;

/*
  Vrzk is vert.x rest+zeebe+kafka
 */
public class VrzkVertx extends AbstractVerticle {

    @Override
    public void start(Promise<Void> startPromise) throws Exception {
        //1.初始化zeebeClient
        ZeebeClient client = ZeebeClient.newClientBuilder()
                .brokerContactPoint("localhost:26500")
                .usePlaintext()
                .build();
        System.out.println("1. ZeebeClient初始化完成");

        //2.部署工作流
        String bpmnPath = "bpmn/vrzk.bpmn";
        client.newDeployCommand()
                .addResourceFromClasspath(bpmnPath)
                .send()
                .join();
        System.out.println("2. vrzk.bpmn部署完成");

        //3.初始化kafka
        final Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("acks", "all");
        props.put("group.id", "GroupA");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.serializer", LongSerializer.class.getName());
        props.put("value.serializer", JsonObjectSerializer.class.getName());
        props.put("key.deserializer", LongDeserializer.class.getName());
        props.put("value.deserializer", JsonObjectDeserializer.class.getName());

        KafkaProducer<Long, JsonObject> producer = KafkaProducer.createShared(vertx, "zeebeOutput", props);
        KafkaConsumer<Long, JsonObject> consumer = KafkaConsumer.create(vertx, props);
        consumer.subscribe("zeebeOutput");
        System.out.println("3. kafka初始化完成");

        //4.部署jobworker,最后一个jobworker传入kafka对象
        String plusType = "plus";
        List<String> plusParam = Arrays.asList("a", "b");
        PlusJobWorkerHandler plusJobWorkerHandler = new PlusJobWorkerHandler(client);
        JobWorker plusJobWorker = client.newWorker()
                .jobType(plusType)
                .handler(plusJobWorkerHandler)
                .fetchVariables(plusParam)
                .open();
        System.out.println("4. plus jobworker部署完成");

        String publishType = "publish";
        String variable = "output";
        PublishJobWorkerHandler publishJobWorkerHandler = new PublishJobWorkerHandler(client, producer);
        JobWorker publishJobWorker = client.newWorker()
                .jobType(publishType)
                .handler(publishJobWorkerHandler)
                .fetchVariables(variable)
                .open();
        System.out.println("5. publish jobworker部署完成");

        //5.创建http服务端
        vertx.createHttpServer()
                .requestHandler(req -> {
                    //6.启动流程
                    String processId = "vrzk";
                    JsonObject input = new JsonObject()
                            .put("a", 4)
                            .put("b", 5);
                    WorkflowInstanceEvent workflowInstanceEvent = client.newCreateInstanceCommand()
                            .bpmnProcessId(processId)
                            .latestVersion()
                            .variables(input.encode())
                            .send()
                            .join();
                    System.out.println("6. 启动vrzk流程成功");

                    //7.kafka拉取消息--使用流程实例key
                    long key = workflowInstanceEvent.getWorkflowInstanceKey();
                    consumer.handler(ar -> {
                        ConsumerRecord<Long, JsonObject> record = ar.record();
                        if (key == record.key()) {
                            System.out.println("7. 工作流实例key=" + key + ",kafka拉取output的topic的消息成功");
                            JsonObject output = record.value();
                            req.response().end(output.encode());
                        }else {
                            System.out.println("消息的key不匹配");
                        }
                    });
                })
                .listen(8888, result -> {
                    if (result.succeeded()) {
                        startPromise.complete();
                    } else {
                        startPromise.fail(result.cause());
                    }
                });
    }
}
